package com.hyw.trace.configuration.meta.redis.queue;

import java.io.IOException;
import java.time.Duration;
import java.util.Objects;
import java.util.function.Consumer;

import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

public abstract class AbstractCacheQueue implements ICacheQueue<String> {

  protected static final Duration timeout = Duration.ofSeconds(2);

  private String topic; // 队列主题

  private volatile boolean running = true;

  protected RedisTemplate<String, String> redisTemplate;

  protected Consumer<String> consumer;

  /**
   * @param topic
   * @param redisTemplate
   * @param consumer
   *          如果为null,则不会从队列中消费任务
   */
  protected AbstractCacheQueue(String topic, RedisTemplate<String, String> redisTemplate,
      @Nullable Consumer<String> consumer) {
    super();
    Assert.isTrue(StringUtils.hasText(topic), "队列名称不能为空");
    Objects.requireNonNull(redisTemplate, "redisTemplate不能为空");

    this.topic = topic;
    this.redisTemplate = redisTemplate;
    this.consumer = consumer;
  }

  public boolean isRunning() {
    return this.running;
  }

  protected abstract void run();

  @Override
  public void handle(String msg) {
    Objects.requireNonNull(consumer);
    consumer.accept(msg);
  }

  @Override
  public boolean add(String msg) {
    Assert.isTrue(running, "队列已经关闭");
    return true;
  }

  @Override
  public void close() throws IOException {
    this.running = false;
  }

  @Override
  public String getTopic() {
    return this.topic;
  }
}
